package com.xian.java.batch;


import com.mongodb.hadoop.io.BSONWritable;
import com.mongodb.hadoop.mapred.MongoInputFormat;
import com.xian.java.MyDataUtil;
import org.apache.flink.api.common.io.RichOutputFormat;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.hadoop.mapred.HadoopInputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapred.JobConf;
import org.bson.BSONObject;

import java.io.IOException;

/**
 * flink获取mongo数据
 */
public class GetMongoData3 {
    public static void main(String[] args) throws Exception {

        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        //String mongoInputUri = parameterTool.get("mongoInputUri");
        //String mongoInputUri = "mongodb://pubuser:KAov0XvKeFPVrF+4+rz-6P49B9U6Rz@172.18.4.25:20017/activity.activityAssistanceRelationPo?authSource=admin";  //指定验证库正确方式
        //String mongoInputUri = "mongodb://pubuser:KAov0XvKeFPVrF+4+rz-6P49B9U6Rz@172.18.4.25:20017/activity.activityAssistanceRelationPo?authSource=activity";  //指定验证库正确方式
        String mongoInputUri = "mongodb://pubuser:KAov0XvKeFPVrF+4+rz-6P49B9U6Rz@172.18.4.25:20017/traffic.ws_traffic_operate_pay?authSource=admin";  //指定验证库正确方式
        System.out.println("传入mongoInputUri="+mongoInputUri);

        // 获取执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 定义输入格式
        HadoopInputFormat<BSONWritable, BSONWritable> hdIf = new HadoopInputFormat<BSONWritable, BSONWritable>(
                new MongoInputFormat(), BSONWritable.class, BSONWritable.class,	new JobConf());

        // 指定mongo 连接url
        hdIf.getJobConf().set("mongo.input.uri", mongoInputUri);

        DataSource<Tuple2<BSONWritable, BSONWritable>> dataSource = env.createInput(hdIf);

        dataSource.print();
        long count = dataSource.count();
        System.out.println("总共数据量为:"+count);


        dataSource.output(new RichOutputFormat<Tuple2<BSONWritable, BSONWritable>>() {

            String zk = "172.18.224.143:2181,172.18.224.144:2181,172.18.224.145:2181";
            Connection con = null;


            @Override
            public void configure(Configuration parameters) {

            }

            @Override
            public void open(int taskNumber, int numTasks) throws IOException {
                org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
                conf.set("hbase.zookeeper.quorum", zk);
                con = ConnectionFactory.createConnection(conf);
                System.out.println("获取到连接:"+con);
            }

            @Override
            public void writeRecord(Tuple2<BSONWritable, BSONWritable> record) throws IOException {
                //System.out.println(record);
                System.out.println("record=="+record);
                System.out.println("record.f1="+record.f1);
                BSONObject doc = record.f1.getDoc();
                //System.out.println("doc="+doc+"\n");
                BSONObject dataDoc = (BSONObject)doc.get("data");
               // System.out.println("dataDoc="+dataDoc);

                String orderId = dataDoc.get("orderId").toString();
                String onArea = dataDoc.get("onArea").toString();
                String orderTime = dataDoc.get("orderTime").toString();
                String startAddr = dataDoc.get("startAddr").toString();
                BSONObject startPoint_Doc = (BSONObject)dataDoc.get("startPoint"); //有嵌套doc  不知道怎么多了 startPoint_Doc={areaCode=440306, lat=22.583963, lon=113.863476, angle=0.0}
                //System.out.println("startPoint_Doc="+startPoint_Doc);
                String startPoint_lat = startPoint_Doc.get("lat").toString();
                String startPoint_lon = startPoint_Doc.get("lon").toString();
                String endAddr = dataDoc.get("endAddr").toString();
                BSONObject endPoint_Doc = (BSONObject)dataDoc.get("endPoint"); //有嵌套doc
                String endPoint_lat = endPoint_Doc.get("lat").toString();
                String endPoint_lon = endPoint_Doc.get("lon").toString();
                String encrypt = dataDoc.get("encrypt").toString();
                BSONObject startVehPoint_Doc = (BSONObject)dataDoc.get("startVehPoint"); //有嵌套doc
                String startVehPoint_speed = startVehPoint_Doc.get("speed").toString();
                String startVehPoint_accelerateSpeed = startVehPoint_Doc.get("accelerateSpeed").toString();
                String startVehPoint_elevation = startVehPoint_Doc.get("elevation").toString();
                String startVehPoint_locTime = startVehPoint_Doc.get("locTime").toString();
                String startVehPoint_accuracy = startVehPoint_Doc.get("accuracy").toString();
                String startVehPoint_address = startVehPoint_Doc.get("address").toString();
                String startVehPoint_locType = startVehPoint_Doc.get("locType").toString();
                String startVehPoint_areaCode = startVehPoint_Doc.get("areaCode").toString();
                String startVehPoint_lat = startVehPoint_Doc.get("lat").toString();
                String startVehPoint_lon = startVehPoint_Doc.get("lon").toString();
                String startVehPoint_angle = startVehPoint_Doc.get("angle").toString();
                BSONObject endVehPoint_Doc = (BSONObject)dataDoc.get("endVehPoint"); //有嵌套doc
                String endVehPoint_speed = endVehPoint_Doc.get("speed").toString();
                String endVehPoint_accelerateSpeed = endVehPoint_Doc.get("accelerateSpeed").toString();
                String endVehPoint_elevation = endVehPoint_Doc.get("elevation").toString();
                String endVehPoint_locTime = endVehPoint_Doc.get("locTime").toString();
                String endVehPoint_accuracy = endVehPoint_Doc.get("accuracy").toString();
                String endVehPoint_address = endVehPoint_Doc.get("address").toString();
                String endVehPoint_locType = endVehPoint_Doc.get("locType").toString();
                String endVehPoint_areaCode = endVehPoint_Doc.get("areaCode").toString();
                String endVehPoint_lat = endVehPoint_Doc.get("lat").toString();
                String endVehPoint_lon = endVehPoint_Doc.get("lon").toString();
                String endVehPoint_angle = endVehPoint_Doc.get("angle").toString();





                //System.out.println("测试值：||"+ endVehPoint_areaCode +"||");


                String rowkey = MyDataUtil.timeStamp2Date(orderTime)+"|"+orderId;
                //System.out.println("rowkey=|"+rowkey+"|");


                TableName tableName = TableName.valueOf("traffic:ws_traffic_operate_pay");
                String cf = "traffic";
                Put put = new Put(Bytes.toBytes(rowkey));
                //添加字段和字段值
                put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("name"),Bytes.toBytes("jemson"));

                /*
                //批量写入提高效率
                List<Put> puts = new ArrayList<>();
                puts.add(put);
                //设置缓存10m，当达到1m时数据会自动刷到hbase
                BufferedMutatorParams bufferedMutatorParams = new BufferedMutatorParams(tableName);
                //设置缓存的大小
                bufferedMutatorParams.writeBufferSize(10 * 1024 * 1024);
                BufferedMutator bufferedMutator = con.getBufferedMutator(bufferedMutatorParams);
                bufferedMutator.mutate(puts);
                bufferedMutator.flush();
                puts.clear();
*/

            }

            @Override
            public void close() throws IOException {

            }
        });







        //务必执行
        env.execute();


        System.out.println("程序运行完毕！");
    }
}
